mod app;
mod args;
mod auth;
mod db;
mod domain;
mod http;
mod infra;
mod notify;
mod storage;
mod sref;

use crate::args::{clap_app, AuthType, FILE_DIR, ITER_COUNTS, LOG_DIR, LOG_KEEP_DAYS, PASSWORD, ROOT_DIR, SUB_COMMAND_USER, USERNAME};
use crate::auth::{DefaultUserProvider, LookupAuthenticator};
use crate::db::init;
use crate::domain::events::{EventDispatcher, FTPEvent};
use crate::domain::user::User;
use crate::infra::userdetail_json::JsonUserProvider;
use crate::notify::FTPListener;
use base64::prelude::BASE64_STANDARD;
use base64::Engine;
use clap::ArgMatches;
use flate2::read::GzDecoder;
use fs4::fs_std::FileExt;
use libnftp::auth::Authenticator;
use libnftp::notification::DataListener;
use libnftp::options;
use libnftp::options::ActivePassiveMode::ActiveAndPassive;
use libnftp::options::{FailedLoginsBlock, FailedLoginsPolicy, SiteMd5};
use libnftp::storage::StorageBackend;
use libnftp::{auth as auth_spi, ServerBuilder};
#[cfg(feature = "pam_auth")]
use nftp_auth_pam as pam;
use nftp_sbe_restrict::RestrictingVfs;
use nftp_sbe_rooter::RooterVfs;
use pbkdf2::pbkdf2_hmac_array;
use rand_core::{OsRng, TryRngCore};
use sha2::Sha256;
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, Write};
use std::net::{IpAddr, SocketAddr};
use std::path::{Path, PathBuf};
use std::process::Command;
use std::str::FromStr;
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use std::{env, fs};
use time::format_description;
use time::macros::offset;
use tokio::runtime;
use tokio::sync::broadcast::Receiver;
use tokio::sync::mpsc::Sender;
use tracing::{info, warn};
use tracing_appender::rolling::{RollingFileAppender, Rotation};
use tracing_subscriber::filter::LevelFilter;
use tracing_subscriber::fmt::time::OffsetTime;

pub const DATE_TIME_FORMAT: &str = "[year]-[month]-[day] [hour]:[minute]:[second].[subsecond digits:3]";

pub static NFTP_INSTANCE_NAME: OnceLock<String> = OnceLock::new();

fn load_user_file(
    path: impl AsRef<Path>,
) -> Result<String, Box<dyn std::error::Error + Send + Sync + 'static>> {
    let mut f = File::open(path)?;

    // The user file can be plaintext, gzipped, or gzipped+base64-encoded
    // The gzip-base64 format is useful for overcoming configmap size limits in Kubernetes
    let mut magic: [u8; 4] = [0; 4];
    let n = f.read(&mut magic[..])?;
    let is_gz = n > 2 && magic[0] == 0x1F && magic[1] == 0x8B && magic[2] == 0x8;
    // the 3 magic bytes translate to "H4sI" in base64
    let is_base64gz =
        n > 3 && magic[0] == b'H' && magic[1] == b'4' && magic[2] == b's' && magic[3] == b'I';

    f.rewind()?;
    if is_gz | is_base64gz {
        let mut gzdata: Vec<u8> = Vec::new();
        if is_base64gz {
            let mut b = Vec::new();
            f.read_to_end(&mut b)?;
            b.retain(|&x| x != b'\n' && x != b'\r');
            gzdata = BASE64_STANDARD.decode(b)?;
        } else {
            f.read_to_end(&mut gzdata)?;
        }
        let mut d = GzDecoder::new(&gzdata[..]);
        let mut s = String::new();
        d.read_to_string(&mut s)?;
        Ok(s)
    } else {
        let mut s = String::new();
        f.read_to_string(&mut s)?;
        Ok(s)
    }
}

type VfsProducer = Box<
    dyn (Fn() -> RooterVfs<
            RestrictingVfs<storage::ChoosingVfs, User, storage::SbeMeta>,
            User,
            storage::SbeMeta,
        >) + Send
        + Sync,
>;

fn start_ftp(
    authenticator: Arc<dyn Authenticator<User> + Send + Sync>,
    arg_matches: ArgMatches,
    shutdown: Receiver<()>,
    done: Sender<()>,
) -> Result<(), String> {
    let event_dispatcher = notify::create_event_dispatcher(&arg_matches)?;
    let svc = |prod: VfsProducer| {
        start_ftp_with_storage(
            &arg_matches,
            prod,
            event_dispatcher,
            authenticator,
            shutdown,
            done,
        )
    };

    match arg_matches
        .get_one::<String>(args::STORAGE_BACKEND_TYPE)
        .map(|backend_type| backend_type.as_str())
    {
        None | Some("filesystem") => svc(fs_storage_backend(&arg_matches)),
        Some(x) => Err(format!("Unknown storage backend type {}", x)),
    }
}

fn start_ftp_with_storage<S>(
    arg_matches: &ArgMatches,
    storage_backend: Box<dyn (Fn() -> S) + Send + Sync>,
    event_dispatcher: Arc<dyn EventDispatcher<FTPEvent>>,
    authenticator: Arc<dyn Authenticator<User> + Send + Sync>,
    mut shutdown: Receiver<()>,
    done: Sender<()>,
) -> Result<(), String>
where
    S: StorageBackend<User> + Send + Sync + 'static,
    S::Metadata: Sync + Send,
{
    let ports = arg_matches
        .get_one::<String>(args::PASSIVE_PORTS)
        .unwrap()
        .split(|c: char| !c.is_numeric())
        .filter(|s| !s.is_empty())
        .collect::<Vec<&str>>();
    if ports.len() != 2 {
        return Err(format!(
            "please specify a valid port range e.g. 50000-60000 for --{}",
            args::PASSIVE_PORTS
        ));
    }

    let start_port: u16 = ports[0]
        .parse()
        .map_err(|_| "start of port range needs to be numeric")?;
    let end_port: u16 = ports[1]
        .parse()
        .map_err(|_| "end of port range needs to be numeric")?;

    let passive_host = get_passive_host_option(arg_matches)?;

    let idle_timeout = arg_matches
        .get_one::<String>(args::IDLE_SESSION_TIMEOUT)
        .unwrap();
    let idle_timeout = String::from(idle_timeout)
        .parse::<u64>()
        .map_err(move |e| {
            format!(
                "unable to parse given value '{}' for --{}: {}. Please use a numeric value",
                idle_timeout,
                args::IDLE_SESSION_TIMEOUT,
                e
            )
        })?;

    let md5_settings = if arg_matches.contains_id(args::ENABLE_SITEMD5) {
        SiteMd5::Accounts
    } else {
        SiteMd5::None
    };

    let hostname = get_host_name();
    let instance_name = arg_matches
        .get_one::<String>(args::INSTANCE_NAME)
        .unwrap()
        .to_owned();

    let listener = Arc::new(FTPListener {
        event_dispatcher: event_dispatcher.clone(),
        instance_name: instance_name.clone(),
        hostname: hostname.clone(),
    });

    let mut server = ServerBuilder::with_authenticator(storage_backend, authenticator)
        .greeting("Welcome to notifyFTP")
        .active_passive_mode(ActiveAndPassive)
        .passive_ports(start_port..end_port)
        .idle_session_timeout(idle_timeout)
        .passive_host(passive_host)
        .sitemd5(md5_settings)
        .notify_data(listener.clone() as Arc<dyn DataListener>)
        .shutdown_indicator(async move {
            shutdown.recv().await.ok();
            options::Shutdown::new().grace_period(Duration::from_secs(10))
        });

    if let Some(arg) = arg_matches.get_one::<String>(args::FAILED_LOGINS_POLICY) {
        let max_attempts = arg_matches
            .get_one::<String>(args::FAILED_MAX_ATTEMPTS)
            .unwrap();
        let max_attempts = max_attempts.parse::<u32>().map_err(move |e| {
            format!(
                "unable to parse given value '{}' for --{}: {}. Please use a numeric value",
                max_attempts,
                args::FAILED_MAX_ATTEMPTS,
                e
            )
        })?;

        let expires_after = arg_matches
            .get_one::<String>(args::FAILED_EXPIRE_AFTER)
            .unwrap();
        let expires_after = expires_after.parse::<u32>().map_err(move |e| {
            format!(
                "unable to parse given value '{}' for --{}: {}. Please use a numeric value",
                expires_after,
                args::FAILED_EXPIRE_AFTER,
                e
            )
        })?;

        let policy = match arg.parse::<args::FailedLoginPolicyType>()? {
            args::FailedLoginPolicyType::ip => FailedLoginsPolicy::new(
                max_attempts,
                Duration::from_secs(expires_after.into()),
                FailedLoginsBlock::IP,
            ),
            args::FailedLoginPolicyType::user => FailedLoginsPolicy::new(
                max_attempts,
                Duration::from_secs(expires_after.into()),
                FailedLoginsBlock::User,
            ),
            args::FailedLoginPolicyType::combination => FailedLoginsPolicy::new(
                max_attempts,
                Duration::from_secs(expires_after.into()),
                FailedLoginsBlock::UserAndIP,
            ),
        };

        server = server.failed_logins_policy(policy);
    }

    let server = server
        .build()
        .map_err(|e| format!("unable to build server: {}", e))?;

    let addr = String::from(arg_matches.get_one::<String>(args::BIND_ADDRESS).unwrap());
    tokio::spawn(async move {
        info!("Starting nftp server at {}", &addr);
        if let Err(e) = server.listen(&addr).await {
            eprintln!("Start nftp at {addr} error, reason: {e:?}");
            std::process::exit(1)
        }
        drop(done)
    });

    Ok(())
}

fn make_auth(
    arg_matches: &ArgMatches,
) -> Result<Arc<dyn Authenticator<User> + Send + Sync + 'static>, String> {
    let default_auth_type = AuthType::Anonymous.to_string();
    let input_auth_type = arg_matches
        .get_one::<String>(args::AUTH_TYPE)
        .unwrap_or(&default_auth_type);
    let auth_type_variant = match input_auth_type.parse::<AuthType>() {
        Ok(t) => t,
        Err(strum::ParseError::VariantNotFound) => {
            return Err(format!("unknown auth type {}", input_auth_type));
        }
    };

    let auth = match auth_type_variant {
        AuthType::Json => make_json_auth(arg_matches),
        AuthType::Anonymous => make_anon_auth(),
        AuthType::Pam => make_pam_auth(arg_matches),
    }?;

    if auth_type_variant != AuthType::Pam && arg_matches.contains_id(args::AUTH_PAM_SERVICE) {
        return Err(format!(
            "parameter {} set while auth_type is set to {}",
            args::AUTH_PAM_SERVICE,
            auth_type_variant
        ));
    }

    Ok(Arc::new(auth))
}

fn make_json_auth(arg_matches: &ArgMatches) -> Result<LookupAuthenticator, String> {
    #[cfg(not(feature = "json_auth"))]
    {
        let _ = arg_matches;
        Err(format!(
            "the jsonfile authentication module was disabled at build time"
        ))
    }
    #[cfg(feature = "json_auth")]
    {
        let path = arg_matches
            .get_one::<String>(ROOT_DIR)
            .unwrap();
        let mut auth_path = PathBuf::from(path);
        auth_path.push("users.json");
        if let Ok(exist) = auth_path.try_exists() {
            if exist {
                let authenticator = nftp_auth_jsonfile::JsonFileAuthenticator::from_file(auth_path.as_path())
                    .map_err(|e| e.to_string())?;

                let mut authenticator = LookupAuthenticator::new(authenticator);
                let json: String = load_user_file(auth_path.as_path())
                    .map_err(|e| format!("Couldn't load user file '{}': {}", path, e))?;
                let user_provider = JsonUserProvider::from_json(json.as_str())?;
                user_provider.users
                    .iter()
                    .filter_map(|u| u.root.as_ref())
                    .for_each(|root| {
                        let mut path = get_store_dir(arg_matches);
                        path.push(root);
                        create_dir(path);
                    });

                authenticator
                    .set_user_detail(Box::new(user_provider));

                return Ok(authenticator)
            }
        }

        eprintln!("Couldn't find users.json in {}", path);
        std::process::exit(1);
    }
}

fn make_pam_auth(arg_matches: &ArgMatches) -> Result<LookupAuthenticator, String> {
    #[cfg(not(feature = "pam_auth"))]
    {
        let _ = arg_matches;
        Err(String::from(
            "the pam authentication module was disabled at build time",
        ))
    }
    #[cfg(feature = "pam_auth")]
    {
        if let Some(service) = arg_matches.get_one::<String>(args::AUTH_PAM_SERVICE) {
            let pam_auth = pam::PamAuthenticator::new(service);
            return Ok(LookupAuthenticator::new(pam_auth));
        }
        Err(format!(
            "--{} is required when using pam auth",
            args::AUTH_PAM_SERVICE
        ))
    }
}

fn make_anon_auth() -> Result<LookupAuthenticator, String> {
    let mut authenticator = LookupAuthenticator::new(auth_spi::AnonymousAuthenticator);
    authenticator.set_user_detail(Box::new(DefaultUserProvider));
    Ok(authenticator)
}

fn get_host_name() -> String {
    if let Ok(host) = env::var("HOST") {
        return host;
    }
    if let Ok(host) = env::var("HOSTNAME") {
        return host;
    }
    match Command::new("hostname").output() {
        Ok(output) => String::from_utf8_lossy(&output.stdout).replace('\n', ""),
        Err(_) => "unknown".to_string(),
    }
}

fn get_passive_host_option(arg_matches: &ArgMatches) -> Result<options::PassiveHost, String> {
    let passive_host = arg_matches
        .get_one::<String>(args::PASSIVE_HOST)
        .map(String::as_str);
    match passive_host {
        None | Some("from-connection") => Ok(options::PassiveHost::FromConnection),
        Some(ip_or_dns) => match ip_or_dns.parse() {
            Ok(IpAddr::V4(ip)) => Ok(options::PassiveHost::Ip(ip)),
            Ok(IpAddr::V6(_)) => Err(format!(
                "an IP is valid for the '--{}' argument, but it needs to be an IP v4 address",
                args::PASSIVE_HOST
            )),
            Err(_) => Err(format!("an IP is valid for the '--{}' argument", ip_or_dns)),
        },
    }
}

fn create_dir<T: AsRef<Path>>(dir: T) {
    let path = dir.as_ref();
    match path.try_exists() {
        Ok(true) => (),
        Ok(false) => {
            if let Err(e) = fs::create_dir(path) {
                eprintln!("Could not create {} dir: {}", path.display(), e);
                std::process::exit(1);
            }
        }
        Err(e) => {
            eprintln!("{e:?}");
            std::process::exit(1);
        }
    }
}

fn get_store_dir(arg_matches: &ArgMatches) -> PathBuf {
    match arg_matches.get_one::<String>(FILE_DIR) {
        Some(dir) => PathBuf::from(dir),
        None => {
            let root = arg_matches.get_one::<String>(ROOT_DIR).unwrap();
            let mut file_dir = PathBuf::from(root);
            file_dir.push("files");
            create_dir(file_dir.as_path());

            file_dir
        }
    }
}

fn fs_storage_backend(arg_matches: &ArgMatches) -> VfsProducer {
    let path = get_store_dir(arg_matches);

    Box::new(move || {
        let fs = nftp_sbe_fs::Filesystem::new(path.clone());
        RooterVfs::new(RestrictingVfs::new(storage::ChoosingVfs {
            inner: storage::InnerVfs::File(fs),
        }))
    })
}

#[derive(PartialEq)]
struct ExitSignal(pub &'static str);

async fn listen_signals() -> Result<ExitSignal, String> {
    #[cfg(unix)]
    {
        use tokio::signal::unix::{signal, SignalKind};
        let mut term_sig = signal(SignalKind::terminate())
            .map_err(|e| format!("Failed to create TERM signal: {}", e))?;
        let mut int_sig = signal(SignalKind::interrupt())
            .map_err(|e| format!("Failed to create INT signal: {}", e))?;
        let mut hup_sig = signal(SignalKind::hangup())
            .map_err(|e| format!("Failed to create HUP signal: {}", e))?;

        let sig_name = tokio::select! {
            Some(_) = term_sig.recv() => "SIG_TERM",
            Some(_) = int_sig.recv() => "SIG_INT",
            Some(_) = hup_sig.recv() => "SIG_HUP",
        };

        Ok(ExitSignal(sig_name))
    }

    #[cfg(windows)]
    {
        use tokio::signal;
        signal::ctrl_c()
            .await
            .map_err(|e| format!("could not listen for ctrl-c: {}", e))?;
        Ok(ExitSignal("CTRL-C"))
    }
}

async fn main_task(arg_matches: ArgMatches) -> Result<ExitSignal, String> {
    let (shutdown_sender, http_receiver) = tokio::sync::broadcast::channel(1);
    let (http_done_sender, mut shutdown_done_received) = tokio::sync::mpsc::channel(1);
    let ftp_done_sender = http_done_sender.clone();

    let ftp_addr: SocketAddr = arg_matches
        .get_one::<String>(args::BIND_ADDRESS)
        .unwrap()
        .parse()
        .map_err(|e| format!("Failed to parse FTP address: {}", e))?;

    let authenticator = make_auth(&arg_matches)?;

    let root_dir = arg_matches
        .get_one::<String>(ROOT_DIR)
        .unwrap();

    // init sqlite
    init(root_dir).map_err(|e| format!("Couldn't init db, reason: {}", e))?;

    let root_dir = String::from(root_dir);
    let ac = authenticator.clone();
    let addr = arg_matches.get_one::<String>(args::HTTP_BIND_ADDRESS).unwrap();
    let addr = String::from(addr);
    tokio::spawn(async move {
        let _ = http::start(
            root_dir,
            ac,
            addr,
            ftp_addr,
            http_receiver,
            http_done_sender,
        )
            .await;
    });


    start_ftp(
        authenticator,
        arg_matches,
        shutdown_sender.subscribe(),
        ftp_done_sender,
    )?;

    let signal = listen_signals().await?;
    drop(shutdown_sender);

    let _ = shutdown_done_received.recv().await;

    Ok(signal)
}

fn run(mut arg_matches: ArgMatches) -> Result<(), String> {
    // init NFTP_INSTANCE_NAME
    NFTP_INSTANCE_NAME.get_or_init(|| {
        let instance_name = arg_matches.get_one::<String>(args::INSTANCE_NAME).unwrap();
        instance_name.into()
    });

    let addr = arg_matches.get_one::<String>(args::BIND_ADDRESS).unwrap();
    let http_addr = arg_matches.get_one::<String>(args::HTTP_BIND_ADDRESS).unwrap();
    let auth_type = arg_matches.get_one::<String>(args::AUTH_TYPE).unwrap();
    let sbe_type = arg_matches.get_one::<String>(args::STORAGE_BACKEND_TYPE).unwrap();

    let home_dir = match sbe_type.as_str() {
        "filesystem" => {
            if arg_matches.contains_id(FILE_DIR) {
                arg_matches.get_one::<String>(FILE_DIR).unwrap()
            } else {
                arg_matches.get_one::<String>(ROOT_DIR).unwrap()
            }
        },
        _ => return Err("Unsupported sbe_type".into()),
    };

    info!(
        "{} server version: {}, ftp-address: {}, http-address: {}, root-dir: {}, auth-type: {}, storage-type: {}.",
        app::NAME,
        app::VERSION,
        addr,
        http_addr,
        home_dir,
        auth_type,
        sbe_type,
    );

    let mut builder = runtime::Builder::new_multi_thread();
    match arg_matches.get_one::<u32>(args::WORKERS) {
        Some(count) => {
            builder.worker_threads(*count as usize);
        },
        None => (),
    };

    let runtime = builder
        .thread_name("nftp-worker")
        .max_blocking_threads(*arg_matches.get_one::<u32>(args::MAX_THREADS).unwrap() as usize)
        .thread_keep_alive(Duration::from_secs(*arg_matches.get_one::<u64>(args::THREAD_KEEPALIVE).unwrap()))
        .enable_all()
        .build()
        .map_err(|err| format!("could not construct runtime {}", err))?;
    while runtime.block_on(main_task(arg_matches))? == ExitSignal("SIG_HUP") {
        arg_matches = clap_app().get_matches();
        warn!("SIG_HUP received, restarting...")
    }

    Ok(())
}

fn lock_directory<P: AsRef<Path>>(path: P) -> std::io::Result<File> {
    let root = path.as_ref();
    let path = path.as_ref();
    if !path.exists() {
        eprintln!("Directory <{}> doesn't exist, please specify a valid location", path.display());
        std::process::exit(1);
    }
    let path = path.join(".lock");
    let lock_file = OpenOptions::new().write(true).create(true).open(path.as_path())?;
    match lock_file.try_lock_exclusive() {
        Ok(acquire_lock) => {
            if !acquire_lock {
                eprintln!("Directory <{}> has been using by another nftp process, please specify a different location", root.display());
                std::process::exit(1);
            } else {
                Ok(lock_file)
            }
        },
        Err(err) => Err(err)
    }
}

fn main() {
    let arg_matches = clap_app().get_matches();
    if let Some(matches) = arg_matches.subcommand_matches(SUB_COMMAND_USER) {
        let username = matches.get_one::<String>(USERNAME).unwrap();
        let password = matches.get_one::<String>(PASSWORD).unwrap();
        let iter_counts = matches.get_one::<u32>(ITER_COUNTS).unwrap();

        let mut salt = [0u8; 8];
        OsRng.try_fill_bytes(&mut salt).unwrap();
        let key = pbkdf2_hmac_array::<Sha256, 32>(password.as_ref(), &salt, *iter_counts);

        let mut stdout = std::io::stdout();
        write!(&mut stdout, "{{\n").unwrap();
        write!(&mut stdout, "    \"username\": {:?},\n", username).unwrap();
        write!(
            &mut stdout,
            "    \"pbkdf2_salt\": {:?},\n",
            BASE64_STANDARD.encode(&salt)
        ).unwrap();
        write!(
            &mut stdout,
            "    \"pbkdf2_key\": {:?},\n",
            BASE64_STANDARD.encode(&key)
        ).unwrap();
        write!(&mut stdout, "    \"pbkdf2_iter\": {},\n", iter_counts).unwrap();
        write!(&mut stdout, "    \"account_enabled\": {},\n", true).unwrap();
        write!(&mut stdout, "    \"root\": {:?}\n", username).unwrap();
        write!(&mut stdout, "}}\n").unwrap();
        std::process::exit(0);
    } else {
        let root_dir = arg_matches.get_one::<String>(ROOT_DIR).unwrap();
        let Ok(_file_lock) = lock_directory(root_dir) else {
            eprintln!("Error occurred in {}", root_dir);
            std::process::exit(1);
        };

        #[cfg(all(unix, not(debug_assertions)))]
        {
            use crate::args::NO_DAEMON;
            if !arg_matches.get_flag(NO_DAEMON) {
                let data_dir = arg_matches
                    .get_one::<String>(ROOT_DIR)
                    .unwrap();
                infra::daemon::daemonize(data_dir, "nftp is running in the background");
            }
        }

        let log_dir = if arg_matches.contains_id(LOG_DIR) {
            let log_dir = arg_matches
                .get_one::<String>(LOG_DIR)
                .unwrap();
            PathBuf::from(log_dir)
        } else {
            let mut log_dir = PathBuf::from(root_dir);
            log_dir.push("logs");
            create_dir(log_dir.as_path());
            log_dir
        };

        let keep_days = arg_matches.get_one::<u32>(LOG_KEEP_DAYS).unwrap();

        let appender = RollingFileAppender::builder()
            .rotation(Rotation::DAILY)
            .filename_prefix("nftp.log")
            .max_log_files(*keep_days as usize)
            .build(log_dir.as_path())
            .expect("could not init log appender");
        let (writer, _guard) = tracing_appender::non_blocking(appender);

        let log_level = arg_matches
            .get_one::<String>(args::LOG_LEVEL)
            .unwrap();

        let offset = offset!(+8);
        let timer = OffsetTime::new(offset, format_description::parse_borrowed::<1>(DATE_TIME_FORMAT).unwrap());
        tracing_subscriber::fmt()
            .with_ansi(false)
            .with_timer(timer)
            .with_writer(writer)
            .with_max_level(LevelFilter::from_str(log_level.as_str()).unwrap())
            .init();

        if let Err(e) = run(arg_matches) {
            eprintln!("\nError: {}", e);
            std::process::exit(1);
        }
    }
}
